package com.shujia;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.List;

public class Demo2API {
    private Connection conn;
    private Admin admin;

    @Before
    public void init() {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");

        try {
            conn = ConnectionFactory.createConnection(conf);
            admin = conn.getAdmin();

        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Test
    public void create_table() throws IOException {
        String name = "tb2";
        HTableDescriptor tb2 = new HTableDescriptor(TableName.valueOf(name));
        HColumnDescriptor cf1 = new HColumnDescriptor("cf1");
        tb2.addFamily(cf1);

        admin.createTable(tb2);

    }

    @Test
    public void drop_table() throws IOException {
        String name = "tb2";
        // 将table设置为disable
        admin.disableTable(TableName.valueOf(name));
        // drop table
        admin.deleteTable(TableName.valueOf(name));
    }

    @Test
    public void put() throws IOException {
        Table tb1 = conn.getTable(TableName.valueOf("tb1"));

        Put put = new Put("000001".getBytes());
        put.addColumn("info".getBytes(), "name".getBytes(), "zhangSan".getBytes());

        tb1.put(put);

    }

    @Test
    public void get() throws IOException {
        Table tb1 = conn.getTable(TableName.valueOf("tb1"));
        Get get = new Get("000001".getBytes());


        Result rs = tb1.get(get);
        byte[] row = rs.getRow();
        System.out.println(Bytes.toString(row));
        byte[] name = rs.getValue("info".getBytes(), "name".getBytes());
        System.out.println(Bytes.toString(name));
    }

    @Test
    public void scan() throws IOException {
        Table tb1 = conn.getTable(TableName.valueOf("User"));
        Scan scan = new Scan();
        // 设置scan的startRow表示从那个rowkey开始，包含本身
        scan.withStartRow("row1".getBytes());
        // 设置scan的endRow表示到哪个rowkey结束，不包含本身
        scan.withStopRow("row4".getBytes());

        // 使用getScanner方法scan table
        ResultScanner scanner = tb1.getScanner(scan);
        Result rs = scanner.next();
        while (rs != null) {
            byte[] row = rs.getRow();
            String rowkey = Bytes.toString(row);
            String column = "";
            if ("row1".equals(rowkey)) {
                column = "name";
            } else if ("row2".equals(rowkey)) {
                column = "age";
            } else if ("row3".equals(rowkey)) {
                column = "sex";
            }
            byte[] value = rs.getValue("info".getBytes(), column.getBytes());
            System.out.println(Bytes.toString(value));
            rs = scanner.next();
        }

    }

    @Test
    public void putAll() throws IOException {
        // 先用tableExists判断表是否存在，不存在则创建，存在则跳过
        if (!admin.tableExists(TableName.valueOf("students"))) {
            HTableDescriptor students = new HTableDescriptor(TableName.valueOf("students"));
            students.addFamily(new HColumnDescriptor("info"));
            admin.createTable(students);
        }

        BufferedReader bufferedReader = new BufferedReader(new FileReader("data/students.txt"));
        String line = bufferedReader.readLine();

        // 获取table
        Table students = conn.getTable(TableName.valueOf("students"));

        while (line != null) {
            // 写HBase
            String[] splits = line.split(",");
            String id = splits[0];
            String name = splits[1];
            int age = Integer.valueOf(splits[2]);
            String gender = splits[3];
            String clazz = splits[4];

            // 以id作为rowkey
            Put put = new Put(id.getBytes());
            put.addColumn("info".getBytes(), "name".getBytes(), name.getBytes());
//            // int 类型的数据无法使用getBytes方法
            put.addColumn("info".getBytes(), "age".getBytes(), Bytes.toBytes(age));
            put.addColumn("info".getBytes(), "gender".getBytes(), gender.getBytes());
            put.addColumn("info".getBytes(), "clazz".getBytes(), clazz.getBytes());

            students.put(put);
            // 读取下一条数据
            line = bufferedReader.readLine();
        }


    }

    @Test
    public void cells() throws IOException {
        Table students = conn.getTable(TableName.valueOf("students"));
        Scan scan = new Scan();
        scan.withStartRow("1500100001".getBytes());
        scan.withStopRow("1500100021".getBytes());
        ResultScanner scanner = students.getScanner(scan);
        for (Result rs : scanner) {
            // 获取rowkey
            String rowkey = Bytes.toString(rs.getRow());
            // 直接通过rs.getValue可能会拿不到数据，因为每条数据它的列不一样
//            rs.getValue("info".getBytes(), "name".getBytes());
//            rs.getValue("info".getBytes(), "age".getBytes());
//            rs.getValue("info".getBytes(), "gender".getBytes());
//            rs.getValue("info".getBytes(), "clazz".getBytes());

            System.out.println("当前行的rowkey为：" + rowkey);
            for (Cell cell : rs.listCells()) {
                String family = Bytes.toString(CellUtil.cloneFamily(cell));
                String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                byte[] valueBytes = CellUtil.cloneValue(cell);
                // 获取rowkey
//                String row = Bytes.toString(CellUtil.cloneRow(cell));

                String value;
                if ("age".equals(qualifier)) {
                    value = String.valueOf(Bytes.toInt(valueBytes));
                } else {
                    value = Bytes.toString(valueBytes);
                }
                System.out.println("当前获取的cell的列簇名为：" + family + ",列名为：" + qualifier + ",cell的值为：" + value);
            }
        }

    }

    @After
    public void close() {
        try {
            if (conn != null) {
                conn.close();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
